#include "MulticastPort.h"
#include "Common.h"
#include "Poco/Observer.h"
#include "Poco/Net/NetException.h"

using Poco::Observer;
using Poco::Net::NetException;
using Poco::Net::ConnectionResetException;
using Poco::Logger;

namespace cm_nat
{
    MulticastPort::MulticastPort(SocketReactor& reactor,ByteBufferPool* pool)
        :m_reactor(reactor),m_buffer_pool(pool),m_stop(false)
    {

    }

    MulticastPort::~MulticastPort()
    {

    }

    int MulticastPort::setup(int type, IPAddress bindIP, SocketAddress groupAddress)
    {
        if( type == MULTICAST_SERVER ){
            m_groupAddress = groupAddress;
            m_socket.bind( SocketAddress(bindIP, m_groupAddress.port()), true);
            m_socket.joinGroup( m_groupAddress.host() );
            m_bindAddress = SocketAddress( bindIP, m_groupAddress.port() );
            
        }else{
            m_socket.bind( SocketAddress(bindIP, 0), true);
            m_bindAddress = m_socket.address();
        }
        poco_debug(g_log_nat, Logger::format("daemon socket address:$0", m_bindAddress.toString()));

        m_reactor.addEventHandler(m_socket, 
            Observer<MulticastPort, ReadableNotification>(*this, &MulticastPort::onReceive) );

        m_thread.start( *this );

        return 0;
    }

    int MulticastPort::teardown()
    {   
        m_stop = true;

        return 0;
    }

    void MulticastPort::registerListener( MulticastPortListener* p_listener )
    {
        m_listener = p_listener;
    }

    SocketAddress& MulticastPort::getBindAddress()
    {
        return m_bindAddress;
    }

    int MulticastPort::send(ByteBuffer* p_data,const SocketAddress& sa)
    {
        if( p_data ){
            SocketAddress* p_sa = new SocketAddress(sa);
            p_data->setData((void*)p_sa);
            p_data->setType(EVENT_TO_SEND_DATA);
            m_queue.enqueueNotification(p_data);
        }
        return 0;
    }

    void MulticastPort::onReceive( ReadableNotification * p_rn)
    {
        p_rn->release();

        ByteBuffer* p_data = m_buffer_pool->get(0);
        if( p_data ){
            p_data->increaseRef();
            SocketAddress* p_sa = new SocketAddress;
            int recv = 0;
            try{
                recv = m_socket.receiveFrom( (void*)p_data->Data(), p_data->Capacity(), *p_sa);
            }
            catch(ConnectionResetException cre) {
                poco_debug(g_log_base, cre.message());
            }
            p_data->setLength(recv);
            p_data->setData(p_sa);
            p_data->setType(EVENT_RECVED_DATA);

            m_queue.enqueueNotification( p_data );
        }

    }

    void MulticastPort::run()
    {
        BaseNotification* p_event = NULL;

        while( !m_stop )
        {
            p_event = (BaseNotification* )m_queue.waitDequeueNotification();

            switch( p_event->getType() )
            {
            case EVENT_RECVED_DATA:
                {
                    ByteBuffer* p_data = (ByteBuffer*)p_event;
                    if( m_listener ){
                        m_listener->onReceive(this, p_data );
                    }else{
                        SocketAddress* p_sa = (SocketAddress*)p_data->getData();
                        delete p_sa;
                        p_data->decreaseRef();
                    }
                }
                break;

            case EVENT_TO_SEND_DATA:
                {
                    ByteBuffer* p_data = (ByteBuffer*)p_event;

                    SocketAddress* p_sa = (SocketAddress*)p_data->getData();
                    poco_debug( g_log_base, Logger::format("UDPPort Send:$0", p_sa->toString()) );

                    try{
                        m_socket.sendTo(p_data->Data(), p_data->Length(), *p_sa);

                    }catch(NetException ne){
                        poco_debug( g_log_base, ne.message() );
                    }

                    delete p_sa;
                    p_data->decreaseRef();
                }
                break;
            default:
                {
                    p_event->release();
                }
                break;
            }
        }

    }

}